package yrm

type Flow[M any] interface {
	Header() any
	MsgChannel() <-chan M
	Err() error
	Cancel()
}

// type ErrorFlow[M any] struct {
// 	Error error
// }

// 请使用Publisher创建Flow
// type Flow[M any] struct {
// 	FlowID     int
// 	Header     []byte
// 	MsgChannel <-chan M
// 	Cancel     func()
// 	errChan    chan error
// 	msgChannel chan<- M
// }

type SimpleFlow[M any] struct {
	DownflowOptions
	msgChannel chan M
	errChan    chan error
	cancel     func()
}

// func NewErrorFlow[M any](err error) *SimpleFlow[M] {
// 	errChan := make(chan error, 1)
// 	errChan <- err
// 	return &SimpleFlow[M]{
// 		errChan: errChan,
// 	}
// }

// func (f *SimpleFlow[M]) FlowId() string {
// 	return fmt.Sprintf("%p", f.errChan)
// }

func (f *SimpleFlow[M]) Header() any {
	return f.DownflowOptions.Header
}

func (f *SimpleFlow[M]) MsgChannel() <-chan M {
	return f.msgChannel
}

func (f *SimpleFlow[M]) Err() (err error) {
	err = <-f.errChan
	f.errChan <- err
	return
}

func (f *SimpleFlow[M]) Cancel() {
	if f.cancel == nil {
		return
	}
	f.cancel()
}

func (f *SimpleFlow[M]) End(err error) {
	e := <-f.errChan
	if e != nil {
		return
	}
	f.errChan <- err
	close(f.msgChannel)
}

func (f *SimpleFlow[M]) Send(msg M) {
	if f.DropMsgIfSlowly && len(f.msgChannel) == f.MsgChanSize {
		//丢弃消息
		return
	}
	f.msgChannel <- msg
}
